package com.lube.codec;

import java.util.List;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.protobuf.InvalidProtocolBufferException;
import com.lube.protobuf.ConnectAckProtos.ConnectAck;
import com.lube.protobuf.ConnectProtos.Connect;
import com.lube.protobuf.PublishAckProtos.PublishAck;
import com.lube.protobuf.PublishProtos.Publish;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

/**
 * 消息解码器
 * <p>
 * 报文协议如下：
 * </p>
 * 
 * <pre>
+--------------+---------+--------------+
|           报文头                           |     数据体               |
+--------------+---------+--------------+
| CommandType  |  Length |    Payload   |
+--------------+---------+--------------+
|      1       |    4    |   ${Length}  |
+--------------+---------+--------------+

1.CommandType：消息指令类型，类型为uint8； 
2.Length：消息体长度，类型为UINT32；
3.Payload: 消息体，使用ProtoBuf压缩。
 * 
 * </pre>
 * 
 * @author caizhiping
 * @date 2016年12月16日 下午4:02:54
 */
public class MessageDecoder extends ByteToMessageDecoder {

	private static final Logger LOG = LogManager.getLogger(MessageDecoder.class);

	/**
	 * 报文头最大长度
	 */
	public static final int MAX_HEADER_BYTES = 5;

	/*
	 * (non-Javadoc)
	 * 
	 * @see io.netty.handler.codec.ByteToMessageDecoder#decode(io.netty.channel.
	 * ChannelHandlerContext, io.netty.buffer.ByteBuf, java.util.List)
	 */
	@Override
	protected void decode(ChannelHandlerContext ctx, final ByteBuf in, List<Object> out) throws Exception {
		int readableBytes = in.readableBytes();

		if (readableBytes < MAX_HEADER_BYTES) {
			return;
		}

		int len = new Long(in.getUnsignedInt(1)).intValue();
		if (readableBytes < len) {
			return;
		}

		try {
			Message msg = doDecode(in);
			if (msg != null) {
				out.add(msg);
			}
		} catch (Exception e) {
			LOG.error(e);
		}

	}

	private Message doDecode(final ByteBuf in) {
		int type = in.readUnsignedByte();
		// 检查消息指令是否正确
		if (type <= 0) {
			return null;
		}

		CommandType commandType = CommandType.valueOf(type);

		switch (commandType) {
		case CONNECT:
			return decodeConnectMessage(in);

		case CONNECTACK:
			return decodeConnectAckMessage(in);

		case PING:
			return decodePingMessage();

		case PINGACK:
			return decodePingAckMessage();

		case PUBLISH:
			return decodePublishMessage(in);

		case PUBACK:
			return decodePubAckMessage(in);

		case DISCONNECT:
			return decodeDisConnectMessage();

		default:
			break;
		}

		throw new IllegalArgumentException("Unknown message type");
	}

	private Message decodeConnectMessage(final ByteBuf in) {

		try {
			int length = new Long(in.readUnsignedInt()).intValue();
			byte[] data = new byte[length];
			in.readBytes(data);

			Connect connect = Connect.parseFrom(data);

			Message msg = new Message();
			msg.setCommandType(CommandType.CONNECT);
			msg.setLength(length);
			msg.setPayload(connect);

			return msg;
		} catch (InvalidProtocolBufferException e) {
			LOG.error("Decode connect message failed:", e);
		}

		return null;
	}

	private Message decodeConnectAckMessage(final ByteBuf in) {
		try {
			int length = new Long(in.readUnsignedInt()).intValue();
			byte[] data = new byte[length];
			in.readBytes(data);

			ConnectAck connectAck = ConnectAck.parseFrom(data);

			Message msg = new Message();
			msg.setCommandType(CommandType.CONNECTACK);
			msg.setLength(length);
			msg.setPayload(connectAck);

			return msg;
		} catch (InvalidProtocolBufferException e) {
			LOG.error("Decode connectAck message failed:", e);
		}

		return null;
	}

	private Message decodePingMessage() {
		Message msg = new Message();
		msg.setCommandType(CommandType.PING);
		msg.setLength(0);
		msg.setPayload(null);

		return msg;
	}

	private Message decodePingAckMessage() {
		Message msg = new Message();
		msg.setCommandType(CommandType.PINGACK);
		msg.setLength(0);
		msg.setPayload(null);

		return msg;
	}

	private Message decodePublishMessage(final ByteBuf in) {
		try {
			int length = new Long(in.readUnsignedInt()).intValue();
			byte[] data = new byte[length];
			in.readBytes(data);

			Publish publish = Publish.parseFrom(data);

			Message msg = new Message();
			msg.setCommandType(CommandType.PUBLISH);
			msg.setLength(length);
			msg.setPayload(publish);

			return msg;
		} catch (InvalidProtocolBufferException e) {
			LOG.error("Decode publish message failed:", e);
		}

		return null;
	}

	private Message decodePubAckMessage(final ByteBuf in) {
		try {
			int length = new Long(in.readUnsignedInt()).intValue();
			byte[] data = new byte[length];
			in.readBytes(data);

			PublishAck pubAck = PublishAck.parseFrom(data);

			Message msg = new Message();
			msg.setCommandType(CommandType.PUBACK);
			msg.setLength(length);
			msg.setPayload(pubAck);

			return msg;
		} catch (InvalidProtocolBufferException e) {
			LOG.error("Decode publishAck message failed:", e);
		}

		return null;
	}

	private Message decodeDisConnectMessage() {
		Message msg = new Message();
		msg.setCommandType(CommandType.DISCONNECT);
		msg.setLength(0);
		msg.setPayload(null);

		return msg;
	}
}
